package cn.zl.springbootmqtt.pub;

import cn.zl.springbootmqtt.util.ymlUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * @Description:MQTT消息推送或订阅客户端
 * @Author: zhanglang
 * @Date: 2020/10/27 14:19
 */
@Slf4j
public class MqttPushClient {
    private MqttClient client;

    private static volatile MqttPushClient mqttPushClient = null;

    public static MqttPushClient getInstance() {
        if (null == mqttPushClient) {
            synchronized (MqttPushClient.class) {
                if (null == mqttPushClient) {
                    mqttPushClient = new MqttPushClient();
                }
            }
        }
        return mqttPushClient;
    }

    private MqttPushClient() {
        connect();
    }

    public void connect() {
        try {
            client = new MqttClient(ymlUtil.MQTT_HOST, "PushServer", new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            //连接选项
            options.setCleanSession(false);
            options.setUserName(ymlUtil.MQTT_USER_NAME);
            options.setPassword(ymlUtil.MQTT_PASSWORD.toCharArray());
            options.setConnectionTimeout(ymlUtil.MQTT_TIMEOUT);
            options.setKeepAliveInterval(ymlUtil.MQTT_KEEP_ALIVE);
            //保留会话
            options.setCleanSession(true);
            //设置回调
            client.setCallback(new PushCallBack());
            //建立连接
            client.connect(options);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布，默认qos为0，非持久化
     *
     * @param topic
     * @param pushMessage
     */
    public void publish(String topic, MqttMessage pushMessage) {
        publish(0, false, topic, pushMessage);
    }

    private void publish(int qos, boolean retained, String topic, MqttMessage pushMessage) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        mqttMessage.setRetained(retained);
        mqttMessage.setPayload(pushMessage.toString().getBytes());

        MqttTopic mqttTopic = client.getTopic(topic);
        if (null == mqttTopic) {
            log.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mqttTopic.publish(mqttMessage);
            token.waitForCompletion();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

}
